package com.whoami.dataplatform.tag_combine;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.nio.ByteBuffer;
import java.util.BitSet;

/**
 * @author tzp
 * @since 2019/8/12
 */
public class ReadTUSeq extends Configured implements Tool {

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();

        String input = args[0];
        int lineToRead = Integer.parseInt(args[1]);

        SequenceFile.Reader in = new SequenceFile.Reader(
                conf,
                SequenceFile.Reader.file(new Path(input))
        );

        BytesWritable key = new BytesWritable();
        BytesWritable value = new BytesWritable();
        Text detailValue = new Text();
        DataOutputBuffer buffer = new DataOutputBuffer();
        SequenceFile.ValueBytes vbytes = in.createValueBytes();

        while (lineToRead > 0) {
            lineToRead--;
            boolean eof = -1 == in.nextRawKey(buffer);
            if (!eof) {
                key.set(buffer.getData(), 0, buffer.getLength());
                buffer.reset();
                in.nextRawValue(vbytes);
                vbytes.writeUncompressedBytes(buffer);
                value.set(buffer.getData(), 0, buffer.getLength());
                buffer.reset();

                System.out.println("------");
                System.out.println(BitSet.valueOf(key.copyBytes()).toString());

                ByteBuffer byteBuffer = ByteBuffer.wrap(value.getBytes());
                int num = byteBuffer.getInt();
                System.out.println(num);

                if (value.getLength() > 4) {
                    detailValue.set(value.getBytes(), 4, value.getLength() - 4);
                    System.out.println(detailValue.toString());
                } else {
                    System.out.println("No details");
                }
            } else {
                System.out.println("eof");
                break;
            }
        }
        in.close();

        return 0;
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new Configuration(), new ReadTUSeq(), args));
    }
}
